package indi.zhifa.study2025.test.rabbitmq.pri.consumer;

import com.alibaba.fastjson2.JSON;
import com.rabbitmq.client.*;
import indi.zhifa.study2025.test.rabbitmq.common.Constant;
import indi.zhifa.study2025.test.rabbitmq.common.entity.OrderMsg;
import indi.zhifa.study2025.test.rabbitmq.common.utils.NWRabbitUtils;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

@Slf4j
public class PriConsumerApplication {

    public static void main(String[] args) throws IOException {
        Connection connection = null;
        try{
            connection = NWRabbitUtils.getConnection();
            if(null == connection){
                return;
            }

            CountDownLatch countDownLatch = new CountDownLatch(2);

            final Connection finalConnection = connection;
            Thread t1 = new Thread(()->{
                try{
                    Channel channelA = finalConnection.createChannel();
                    Map<String,Object> param = new HashMap<>();
                    param.put("x-priority",1);
                    // 消息预加载
                    channelA.basicQos(5);
                    channelA.basicConsume(Constant.NW_QUE_PRI_NAME,false,param,new DefaultConsumer(channelA){

                        @SneakyThrows
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                            String str = new String(body);
                            OrderMsg orderMsg = JSON.parseObject(str,OrderMsg.class);
                            log.info("channelA#接收到订单信息："+orderMsg);
                            Thread.sleep(1000);
                            channelA.basicAck(envelope.getDeliveryTag(),false);
                        }
                    });
                    countDownLatch.countDown();
                }catch (Exception e){
                    System.out.println("channelA error:"+e);
                }
            });

            Thread t2 = new Thread(()->{
                try{
                    Channel channelB = finalConnection.createChannel();
                    Map<String,Object> param = new HashMap<>();
                    param.put("x-priority",2);
                    // 消息预加载
                    channelB.basicQos(10);
                    channelB.basicConsume(Constant.NW_QUE_PRI_NAME,false,param,new DefaultConsumer(channelB){

                        @SneakyThrows
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                            String str = new String(body);
                            OrderMsg orderMsg = JSON.parseObject(str,OrderMsg.class);
                            log.info("channelB#接收到订单信息："+orderMsg);
                            Thread.sleep(1000);
                            channelB.basicAck(envelope.getDeliveryTag(),false);
                        }
                    });
                    countDownLatch.countDown();
                }catch (Exception e){
                    System.out.println("channelB error:"+e);
                }
            });

            t1.start();
            t2.start();

            countDownLatch.await();

        }catch (Exception e){
            System.out.println(e);
        }
    }


}
